package com.jiayuan.cn.energy.flink.redis;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
//import org.junit.Test;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import java.util.LinkedHashMap;

public class RedisSourceFunction {

    public static void main(String[] args) {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        Collection<String > cs = new ArrayList<String>();
//        cs.add("Alice");
//        cs.add("Bob");
//        cs.add("Twn");
//        DataStreamSource<String> stream  = env.fromData(cs);
            DataStream<Row> ds = env.fromData(getTestData()).returns(getTestDataType());
            tableEnv.createTemporaryView("order_info", ds);

            tableEnv.executeSql(" CREATE TABLE datasink (item_id string,price String) WITH (" +
                    "    'connector.type' = 'redis'," +
                    "    'connector.cluster.ip' = 'localhost'," +
                    "    'connector.cluster.port' = '6379'," +
                    "    'connector.max.timeout.millis' = '1000'," +
                    "    'connector.max.total' = '1'," +
                    "    'connector.max.idle' = '1'," +
                    "    'connector.min.idle' = '1'," +
                    "    'connector.data.type' = 'string'," +
                    "    'connector.expire.second' = '600'" +
                    ")");
            tableEnv.executeSql("insert into datasink SELECT CONCAT('string',item_id) ,cast(price as string) FROM order_info");
//            env.execute("test");
        }

        private static List<Row> getTestData() {
            List<Row> data = new ArrayList<>();
            data.add(Row.of("20200312", "1", 13.2D, 2L));
            data.add(Row.of("20200312", "2", 12.2D, 2L));
            data.add(Row.of("20200312", "3", 11.2D, 2L));
            data.add(Row.of("20200312", "3", 21.2D, 2L));
            return data;
        }

        private static RowTypeInfo getTestDataType() {
            TypeInformation<?>[] types = {
                    BasicTypeInfo.STRING_TYPE_INFO,
                    BasicTypeInfo.STRING_TYPE_INFO,
                    BasicTypeInfo.DOUBLE_TYPE_INFO,
                    BasicTypeInfo.LONG_TYPE_INFO};
            String[] names = {"pay_hour", "item_id", "price", "total_count"};

            RowTypeInfo typeInfo = new RowTypeInfo(types, names);
            return typeInfo;
        }
}